package com.mall.order.rocketmq;

import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import com.mall.order.model.MqMessage;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by IntelliJ IDEA.
 *
 * @Author: likelu
 * @Date: 2021/4/29 17:34
 * @Description:
 *      发送异步消息
 *       异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。
 */
@Configuration
@Component
public class RobackProducer {

    public String rocketAddr = "106.14.33.182:9876";

    public void msgToMq() throws Exception {

        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name_roback");
        // 设置NameServer的地址
        producer.setNamesrvAddr(rocketAddr);
        // 启动Producer实例
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);

        int messageCount = 1;
        // 根据消息数量实例化倒计时计算器
        final CountDownLatch2 countDownLatch = new CountDownLatch2(messageCount);
        for (int i = 0; i < messageCount; i++) {

            final int index = i;
            // 创建消息，并指定Topic，Tag和消息体
//            Message msg = new Message("TopicTest",
//                    "TagA",
//                    "OrderID188",
//                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
            MqMessage mm = new MqMessage();
            mm.setTopic("TopicOrderRoback");
            mm.setTags("Roback");
            mm.setKeys("OrderID188");
            mm.setBody("A001".getBytes(RemotingHelper.DEFAULT_CHARSET));
            //Thread.sleep(5000);
            // SendCallback接收异步返回结果的回调
            producer.send(mm, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                }
                @Override
                public void onException(Throwable e) {
                    e.printStackTrace();
                    System.out.printf("出现异常");
                }
            });
        }

        // 等待5s
        countDownLatch.await(5, TimeUnit.SECONDS);
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }

}
